package kafka

import (
	"cmdTools/common"
	"container/ring"
	"context"
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"go.uber.org/zap"
	"gopkg.in/yaml.v3"
	"os"
	"os/signal"
	"strconv"
	"sync"
	"syscall"
	"time"
)

func (caHandler *Kafkaer) CreateClusterAdmin() {

	config := sarama.NewConfig()
	config.Version = sarama.V2_8_1_0
	config.Metadata.Timeout = 30 * time.Second
	config.Net.KeepAlive = 10 * time.Second

	caAdmin, err := sarama.NewClusterAdmin(caHandler.brokerList, config)

	if err != nil {
		ZapLog.Error("CreateClusterAdmin",
			zap.Any("Brokers", caHandler.brokerList),
			zap.Any("err", err),
		)
		return
	}

	caHandler.caAdmin = caAdmin
	//go caHandler.checkAdmin()
}

func (caHandler *Kafkaer) GetAllTopic() map[string]sarama.TopicDetail {
	var err error
	caHandler.topicDetail, err = caHandler.caAdmin.ListTopics()
	if err != nil {
		ZapLog.Error("GetAllTopic",
			zap.Any("Status", "failed"),
			zap.Any("error", err),
		)
		return nil
	}
	return caHandler.topicDetail
}

func (caHandler *Kafkaer) GetAllBorkers() {
	brokers, controllerID, err := caHandler.caAdmin.DescribeCluster()
	if err != nil {
		ZapLog.Error("GetAllBorkers",
			zap.Any("Status", "failed"),
			zap.Any("error", err),
		)
	}
	caHandler.broker = brokers
	caHandler.controllerID = controllerID
	return
}

func (caHandler *Kafkaer) ExpandPartitionsForAll() {
	caHandler.GetAllBorkers()
	brokerTotal := len(caHandler.broker)

	for topicName := range caHandler.GetAllTopic() {
		if topicName != "__consumer_offsets" {
			topicMeta, topicMetaErr := caHandler.caAdmin.DescribeTopics([]string{topicName})
			if topicMetaErr != nil {
				ZapLog.Error("ExpandPartitionsForAll",
					zap.Any("topicName", topicName),
					zap.Any("error", topicMetaErr),
				)
				continue
			}

			if len(topicMeta[0].Partitions) < brokerTotal {

				cpErr := caHandler.caAdmin.CreatePartitions(topicName, int32(brokerTotal), nil, false)
				if cpErr != nil {
					ZapLog.Error("ExpandPartitionsForAll",
						zap.Any("topicName", topicName),
						zap.Any("error", cpErr),
					)
				}
			}

			assignments := caHandler.createAssignmentsPlan(topicName)
			err := caHandler.caAdmin.AlterPartitionReassignments(topicName, assignments)
			if err != nil {
				ZapLog.Error("AlterPartitionReassignments",
					zap.Any("TopicName", topicName),
					zap.Any("Error", err),
				)
			}

			mapToYaml(map[string]string{
				"type":      "expandpartitionsforall",
				"status":    "succeed",
				"topicName": topicName,
			})
		}
	}

	return
}

func (caHandler *Kafkaer) CreatePartitions(topic, ra string, np int) interface{} {

	var Assignment [][]int32
	if jsonUnmarshalErr := json.Unmarshal([]byte(ra), &Assignment); jsonUnmarshalErr != nil {
		ZapLog.Error("CreatePartitions",
			zap.Any("TopicName", topic),
			zap.Any("Assignment", ra),
			zap.Any("Error", jsonUnmarshalErr),
		)
		return map[string]string{
			"type":       "CreatePartitions",
			"status":     "failed",
			"topicName":  topic,
			"Assignment": ra,
			"error":      jsonUnmarshalErr.Error(),
		}
	}

	CreatePartitionsErr := caHandler.caAdmin.CreatePartitions(topic, int32(np), Assignment, false)
	if CreatePartitionsErr != nil {
		ZapLog.Error("CreatePartitionsFailed",
			zap.Any("TopicName", topic),
			zap.Any("Partitions", np),
			zap.Any("Error", CreatePartitionsErr),
		)
		return map[string]string{
			"type":       "CreatePartitions",
			"status":     "failed",
			"topicName":  topic,
			"partitions": strconv.Itoa(np),
			"error":      CreatePartitionsErr.Error(),
		}
	}

	result := map[string]string{
		"type":       "CreatePartitions",
		"status":     "succeed",
		"topicName":  topic,
		"partitions": strconv.Itoa(np),
	}
	mapToYaml(result)
	return result

}

func (caHandler *Kafkaer) UpdatePartitionReassignments(topic string, ra string) interface{} {

	var Assignment [][]int32
	if err := json.Unmarshal([]byte(ra), &Assignment); err != nil {
		ZapLog.Error("UpdatePartitionReassignments",
			zap.Any("TopicName", topic),
			zap.Any("Assignment", ra),
			zap.Any("Error", err),
		)
		return map[string]string{
			"type":       "UpdatePartitionReassignments",
			"status":     "failed",
			"topicName":  topic,
			"Assignment": ra,
			"error":      err.Error(),
		}
	}

	if len(Assignment) == 0 {
		Assignment = caHandler.createAssignmentsPlan(topic)
	}

	err := caHandler.caAdmin.AlterPartitionReassignments(topic, Assignment)
	if err != nil {
		ZapLog.Error("UpdatePartitionReassignments",
			zap.Any("TopicName", topic),
			zap.Any("Assignment", ra),
			zap.Any("Error", err),
		)
		return map[string]string{
			"type":       "UpdatePartitionReassignments",
			"status":     "failed",
			"topicName":  topic,
			"Assignment": ra,
			"error":      err.Error(),
		}
	}

	result := map[string]string{
		"type":       "UpdatePartitionReassignments",
		"status":     "succeed",
		"topicName":  topic,
		"assignment": fmt.Sprintf("%v", Assignment),
	}
	//mapToJson(result)
	mapToYaml(result)
	return result
}

func (caHandler *Kafkaer) createAssignmentsPlan(topicName string) (Assignment [][]int32) {

	caHandler.GetAllBorkers()
	brokerTotal := len(caHandler.broker)

	brokerRing := ring.New(brokerTotal)
	for i := int32(0); i < int32(brokerTotal); i++ {
		brokerRing.Value = i
		brokerRing = brokerRing.Next()
	}

	topicMeta, err := caHandler.caAdmin.DescribeTopics([]string{topicName})
	if err != nil {
		ZapLog.Error("createAssignmentsPlan",
			zap.Any("TopicName", topicName),
			zap.Any("Error", err),
		)
		return nil
	}

	Assignment = make([][]int32, len(topicMeta[0].Partitions))
	for i := range Assignment {
		Assignment[i] = make([]int32, len(topicMeta[0].Partitions[0].Replicas))
	}

	for partiton := int32(0); partiton < int32(len(topicMeta[0].Partitions)); partiton++ {

		replicas := len(topicMeta[0].Partitions[0].Replicas)
		replicasSlice := make([]int32, len(topicMeta[0].Partitions[0].Replicas))

		for rf := 0; rf < replicas; rf++ {
			replicasSlice[rf] = brokerRing.Value.(int32)
			brokerRing = brokerRing.Next()
		}

		Assignment[partiton] = replicasSlice
	}

	return Assignment
}

func (caHandler *Kafkaer) CreateClient() {

	config := sarama.NewConfig()
	config.Version = sarama.V2_8_1_0
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	config.ChannelBufferSize = 1000
	config.Net.KeepAlive = 0

	var err error = nil

	caHandler.client, err = sarama.NewClient(caHandler.brokerList, config)

	if err != nil {
		ZapLog.Error("CreateClient",
			zap.Any("Error", err),
		)
	}

	//go caHandler.checkClient()

	return
}

/*func (caHandler *Kafkaer) checkAdmin() {
	go func(caHandler *Kafkaer) {
		for {
			for l, _ := range caHandler.caAdmin {
				_, err := caHandler.caAdmin[l].ListTopics()
				if err != nil {
					ZapLog.Error("checkAdmin",
						zap.String("KafkaLabel", l),
						zap.String("Error", err.Error()),
					)
				}
				/*	ZapLog.Info("checkAdmin",
					zap.String("KafkaLabel", l),
					zap.String("Status", "OK"),
				)
			}
			time.Sleep(10 * time.Second)
		}
	}(caHandler)
}*/

/*
	func (caHandler *Kafkaer) checkClient() {
		go func(caHandler *Kafkaer) {
			for {
				for l, _ := range caHandler.client {
					_, err := caHandler.client[l].RefreshController()
					if err != nil {
						ZapLog.Error("checkClient",
							zap.String("KafkaLabel", l),
							zap.String("Error", err.Error()),
						)
					}
						ZapLog.Info("checkClient",
						zap.String("KafkaLabel", l),
						zap.String("Status", "OK"),
					)
				}
				time.Sleep(10 * time.Second)
			}
		}(caHandler)
	}
*/
func (caHandler *Kafkaer) DescribeTopic(topic []string) interface{} {
	topicMeta, topicMetaErr := caHandler.caAdmin.DescribeTopics(topic)
	if topicMetaErr != nil {
		ZapLog.Error("DescribeTopic",
			zap.Any("TopicName", topic),
			zap.Any("Error", topicMetaErr),
		)
		return map[string]string{
			"type":      "CreateTopic",
			"status":    "failed",
			"topicName": fmt.Sprintf("%v", topic),
			"error":     topicMetaErr.Error(),
		}
	}

	var result []interface{}
	for _, t := range topicMeta {
		topicMap := make(map[string]interface{})
		topicMap["TopicName"] = t.Name
		partitionDesc := make([]interface{}, 0)

		for _, p := range t.Partitions {
			OffsetNewest, OffsetNewestErr := caHandler.client.GetOffset(t.Name, p.ID, sarama.OffsetNewest)
			if OffsetNewestErr != nil {
				ZapLog.Error("DescribeTopic",
					zap.Any("TopicName", t.Name),
					zap.Any("Partitions", p.ID),
					zap.Any("Error", OffsetNewestErr),
				)
				OffsetNewest = 0
			}
			offsetOldest, offsetOldestErr := caHandler.client.GetOffset(t.Name, p.ID, sarama.OffsetOldest)
			if OffsetNewestErr != nil {
				ZapLog.Error("DescribeTopic",
					zap.Any("TopicName", t.Name),
					zap.Any("Partitions", p.ID),
					zap.Any("Error", offsetOldestErr),
				)
				offsetOldest = 0
			}

			partitionDesc = append(partitionDesc, map[string]string{
				"Id":           strconv.Itoa(int(p.ID)),
				"Replicas":     fmt.Sprintf("%v", p.Replicas),
				"Isr":          fmt.Sprintf("%v", p.Isr),
				"Leader":       strconv.Itoa(int(p.Leader)),
				"OffsetNewest": strconv.Itoa(int(OffsetNewest)),
				"offsetOldest": strconv.Itoa(int(offsetOldest)),
			})
		}

		topicMap["partition"] = partitionDesc
		result = append(result, topicMap)
	}

	//mapToJson(result)
	mapToYaml(result)
	return result
}

func (caHandler *Kafkaer) DescribeConsumerGroup(consumerGroup []string) interface{} {

	if len(consumerGroup) == 0 {
		cg, err := caHandler.caAdmin.ListConsumerGroups()
		if err != nil {
			ZapLog.Error("ListConsumerGroups",
				zap.Any("Error", err),
			)
			return map[string]string{
				"type":   "DescribeConsumerGroup",
				"status": "failed",
				"stage":  "ListConsumerGroups",
				"error":  err.Error(),
			}
		}

		for k := range cg {
			consumerGroup = append(consumerGroup, k)
		}

	}

	consumerGroupDescription, err := caHandler.caAdmin.DescribeConsumerGroups(consumerGroup)

	if err != nil {
		ZapLog.Error("DescribeConsumerGroup",
			zap.Any("ConsumerGroup", consumerGroupDescription),
			zap.Any("Error", err),
		)
		return map[string]string{
			"type":          "DescribeConsumerGroup",
			"status":        "failed",
			"consumerGroup": fmt.Sprintf("%s", consumerGroup),
			"error":         err.Error(),
		}
	}

	consumerGroupTotal := len(consumerGroupDescription)
	var result []interface{}
	for cgIndex := 0; cgIndex < consumerGroupTotal; cgIndex++ {
		groupId := consumerGroupDescription[cgIndex].GroupId
		//state := consumerGroupDescription[cgIndex].State
		members := consumerGroupDescription[cgIndex].Members

		consumerGroupMap := make(map[string]interface{})

		memberMap := make([]map[string]interface{}, 0)
		//memberMap := make(map[string]map[string]interface{})
		//memberMap := make(map[int]map[string]interface{})

		for member := range members {
			var lagTotal int64 = 0
			consumerGroupMemberAssignment, _ := consumerGroupDescription[cgIndex].Members[member].GetMemberAssignment()
			OffsetFetchResponse, _ := caHandler.caAdmin.ListConsumerGroupOffsets(groupId, consumerGroupMemberAssignment.Topics)

			topicMap := make(map[string]interface{})
			for topicName, blockValue := range OffsetFetchResponse.Blocks {

				var blocksSlice []interface{}
				for i, k := range blockValue {
					OffsetNewest, OffsetNewestErr := caHandler.client.GetOffset(topicName, i, sarama.OffsetNewest)

					if OffsetNewestErr != nil {
						ZapLog.Error("DescribeTopic",
							zap.Any("TopicName", topicName),
							zap.Any("Partitions", i),
							zap.Any("Error", OffsetNewestErr),
						)
						OffsetNewest = 0
					}

					var lag int64
					if k.Offset == -1 {
						lag = 0
					} else {
						lag = OffsetNewest - k.Offset
					}
					lagTotal += lag
					blocksSlice = append(blocksSlice, map[string]string{
						"PartitionId":  strconv.Itoa(int(i)),
						"Offset":       strconv.Itoa(int(k.Offset)),
						"LeaderEpoch":  strconv.Itoa(int(k.LeaderEpoch)),
						"Lag":          strconv.Itoa(int(lag)),
						"OffsetNewest": strconv.Itoa(int(OffsetNewest)),
					})
				}
				topicMap[topicName] = blocksSlice

			}

			tmpMap := make(map[string]interface{})
			tmpMap["MemberId"] = consumerGroupDescription[cgIndex].Members[member].MemberId
			tmpMap["ClientIP"] = consumerGroupDescription[cgIndex].Members[member].ClientHost[1:]
			tmpMap["Topic"] = topicMap
			tmpMap["LagTotal"] = lagTotal
			memberMap = append(memberMap, tmpMap)

		}

		consumerGroupMap["GroupId"] = groupId
		consumerGroupMap["Members"] = memberMap
		result = append(result, consumerGroupMap)
	}
	//mapToJson(result)
	mapToYaml(result)
	return result
}

func mapToJson(m interface{}) {
	jsonTxt, err := json.Marshal(m)
	if err != nil {
		ZapLog.Error("DescribeTopic",
			zap.Any("Error", err),
		)
	}
	fmt.Println(string(jsonTxt))
}

func mapToYaml(m interface{}) {
	yamlResult, err := yaml.Marshal(m)
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println(string(yamlResult))
}

func (caHandler *Kafkaer) ConsumerClient(topic []string, beging bool, startTimes string) func() {
	caHandler.cg = ConsumerGroups{
		startOffset:       beging,
		version:           "2.8.0",
		ready:             make(chan bool),
		group:             "cmdToolsGroup",
		channelBufferSize: 1000,
		assignor:          "range",
		client:            nil,
		startTime:         startTimes,
	}

	config := sarama.NewConfig()
	version, _ := sarama.ParseKafkaVersion(caHandler.cg.version)
	config.Version = version
	config.Consumer.Group.Rebalance.GroupStrategies = nil
	config.ChannelBufferSize = caHandler.cg.channelBufferSize

	caHandler.cg.client, _ = sarama.NewConsumerGroupFromClient(caHandler.cg.group, caHandler.client)

	ctx, cancel := context.WithCancel(context.Background())
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			if err := caHandler.cg.client.Consume(ctx, topic, caHandler); err != nil {
				// 当setup失败的时候，error会返回到这里
				fmt.Println("Error from consumer: %v", err)
				return
			}
			// check if context was cancelled, signaling that the consumer should stop
			if ctx.Err() != nil {
				fmt.Println(ctx.Err())
				return
			}
			caHandler.cg.ready = make(chan bool)
		}
	}()
	<-caHandler.cg.ready
	fmt.Println("cmdTools consumer up and running!...")

	return func() {
		cancel()
		wg.Wait()
		if err := caHandler.cg.client.Close(); err != nil {
			fmt.Println("Error closing client: %v", err)
		}
	}
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (caHandler *Kafkaer) Setup(session sarama.ConsumerGroupSession) error {
	if caHandler.cg.startOffset {
		for topic, partition := range session.Claims() {
			for _, p := range partition {
				offset, _ := caHandler.client.GetOffset(topic, p, sarama.OffsetOldest)
				session.MarkOffset(topic, p, offset, "")
				session.ResetOffset(topic, p, offset, "")
			}
		}
	}
	fmt.Println(len(caHandler.cg.startTime), " ", caHandler.cg.startTime)
	if len(caHandler.cg.startTime) != 0 {
		startTime, _ := time.Parse("2006-01-02 15:04:05.000", caHandler.cg.startTime)
		println("utc: ", startTime.UnixMilli())
		for topic, partition := range session.Claims() {
			for _, p := range partition {
				offset, _ := caHandler.client.GetOffset(topic, p, startTime.UnixMilli())
				fmt.Println("offset:   ", offset)
				session.MarkOffset(topic, p, offset, "")
				session.ResetOffset(topic, p, offset, "")
			}
		}
	}

	fmt.Println(session.Claims())
	// Mark the consumer as ready
	close(caHandler.cg.ready)
	return nil

}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (caHandler *Kafkaer) Cleanup(sarama.ConsumerGroupSession) error {
	fmt.Println("cleanup")
	//caHandler.cg.client.Close()
	//caHandler.client.Close()
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (caHandler *Kafkaer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// <https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29>
	// 具体消费消息
	for message := range claim.Messages() {
		fmt.Printf("[topic:%s] [partiton:%d] [offset:%d] [time:%v] [key:%s] [value:%s]\n",
			message.Topic, message.Partition, message.Offset, message.Timestamp, message.Key, message.Value)
		session.MarkMessage(message, "")

	}

	return nil
}

func (caHandler *Kafkaer) ConsumerTopic(topics []string, beging bool, startTime string) {
	c := caHandler.ConsumerClient(topics, beging, startTime)

	sigterm := make(chan os.Signal, 1)
	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
	select {
	case <-sigterm:
		fmt.Println("terminating: via signal")
	}
	c()
}
func (caHandler *Kafkaer) DeleteTopic(topic string) interface{} {

	deleteTopicErr := caHandler.caAdmin.DeleteTopic(topic)
	if deleteTopicErr != nil {
		ZapLog.Error("deleteTopicFailed",
			zap.Any("TopicName", topic),
			zap.Any("Error", deleteTopicErr),
		)
		return map[string]string{
			"type":      "CreateTopic",
			"status":    "failed",
			"topicName": topic,
			"error":     deleteTopicErr.Error(),
		}
	}

	result := map[string]string{
		"type":      "DeleteTopic",
		"status":    "succeed",
		"topicName": topic,
	}
	mapToYaml(result)
	return result
}

func (caHandler *Kafkaer) CreateTopic(topic string, np, rc int) interface{} {

	topicDetail := sarama.TopicDetail{
		NumPartitions:     int32(np),
		ReplicationFactor: int16(rc),
		ReplicaAssignment: nil,
		ConfigEntries:     nil,
	}

	createTopicErr := caHandler.caAdmin.CreateTopic(topic, &topicDetail, false)
	if createTopicErr != nil {
		ZapLog.Error("createTopicFailed",
			zap.Any("TopicName", topic),
			zap.Any("Partitions", np),
			zap.Any("Error", createTopicErr),
		)
		return map[string]string{
			"type":       "CreateTopic",
			"status":     "failed",
			"topicName":  topic,
			"partitions": strconv.Itoa(np),
			"error":      createTopicErr.Error(),
		}
	}

	result := map[string]string{
		"type":       "CreateTopic",
		"status":     "succeed",
		"topicName":  topic,
		"partitions": strconv.Itoa(np),
	}
	mapToYaml(result)
	return result
}

func (caHandler *Kafkaer) SendMsgToTopic(topic string, key string, value string) interface{} {

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForLocal
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true

	msg := &sarama.ProducerMessage{
		Topic:     topic,
		Key:       sarama.StringEncoder(key),
		Value:     sarama.StringEncoder(value),
		Headers:   nil,
		Metadata:  nil,
		Timestamp: time.Now(),
	}

	client, err := sarama.NewSyncProducer(caHandler.brokerList, config)
	if err != nil {
		ZapLog.Info("SendMsgToTopic",
			zap.Any("TopicName", topic),
			zap.Any("error", err.Error()),
		)

		return map[string]string{
			"type":      "SendMsgToTopic",
			"status":    "failed",
			"topicName": topic,
			"error":     err.Error(),
		}
	}
	defer func(client sarama.SyncProducer) {
		err := client.Close()
		if err != nil {
		}
	}(client)

	pid, offset, err := client.SendMessage(msg)
	if err != nil {
		ZapLog.Info("SendMsgToTopic",
			zap.Any("TopicName", topic),
			zap.Any("Partitions", pid),
			zap.Any("offset", offset),
			zap.Any("error", err.Error()),
		)

		return map[string]string{
			"type":       "SendMsgToTopic",
			"status":     "failed",
			"topicName":  topic,
			"partitions": strconv.Itoa(int(pid)),
			"offset":     strconv.Itoa(int(offset)),
			"error":      err.Error(),
		}
	}
	result := map[string]string{
		"type":       "SendMsgToTopic",
		"status":     "succeed",
		"topicName":  topic,
		"partitions": strconv.Itoa(int(pid)),
		"offset":     strconv.Itoa(int(offset)),
	}
	mapToYaml(result)

	return result

}

func (caHandler *Kafkaer) KafkaInit(kafkaAddr []string) {

	caHandler.brokerList = kafkaAddr
	zapConfig := zap.Config{
		Level:            common.Atom,
		Development:      false,
		Encoding:         "console",
		EncoderConfig:    common.EncoderConfig,
		InitialFields:    map[string]interface{}{"serviceName": "Kafka"},
		OutputPaths:      []string{"stdout"},
		ErrorOutputPaths: []string{"stderr"},
	}

	var err error
	ZapLog, err = zapConfig.Build()
	if err != nil {
		fmt.Println("zaplog ", err)
	}

}
